#include <string>

#include "common/exception.h"
#include "common/logger.h"
#include "common/rid.h"
#include "storage/index/b_plus_tree.h"
#include "storage/page/header_page.h"

int i = 1;

namespace bustub {
INDEX_TEMPLATE_ARGUMENTS
BPLUSTREE_TYPE::BPlusTree(std::string name, BufferPoolManager *buffer_pool_manager, const KeyComparator &comparator,
                          int leaf_max_size, int internal_max_size)
    : index_name_(std::move(name)),
      root_page_id_(INVALID_PAGE_ID),
      buffer_pool_manager_(buffer_pool_manager),
      comparator_(comparator),
      leaf_max_size_(leaf_max_size),
      internal_max_size_(internal_max_size) {}

/*
 * Helper function to decide whether current b+tree is empty
 */
INDEX_TEMPLATE_ARGUMENTS
auto BPLUSTREE_TYPE::IsEmpty() const -> bool { return root_page_id_ == INVALID_PAGE_ID; }
/*****************************************************************************
 * SEARCH
 *****************************************************************************/
/*
 * Return the only value that associated with input key
 * This method is used for point query
 * @return : true means key exists
 */
INDEX_TEMPLATE_ARGUMENTS
auto BPLUSTREE_TYPE::GetValue(const KeyType &key, std::vector<ValueType> *result, Transaction *transaction) -> bool {
  // std::scoped_lock<std::mutex> lock(latch_);
  // 只取了leaf_page，读完后释放读锁和Unpin即可
  root_latch_.RLock();
  LeafPage *leaf_page = GetLeafPage(Operation::Read,key,transaction);
  if (leaf_page == nullptr) {
    root_latch_.RUnlock();
    return false;
  }
  ValueType value;
  bool is_found = leaf_page->FindKey(key, value, comparator_);
  if (is_found) {
    result->emplace_back(value);
  }
  auto page_set = transaction->GetPageSet();
  page_set->back()->RUnlatch();
  page_set->pop_back();
  buffer_pool_manager_->UnpinPage(leaf_page->GetPageId(), false);
  return is_found;
}

INDEX_TEMPLATE_ARGUMENTS
auto BPLUSTREE_TYPE::IsSafePage(Page *page, Operation op) -> bool {
  if (op == Operation::Read) {
    return true;
  } 
  BPlusTreePage *tree_page = reinterpret_cast<BPlusTreePage *>(page->GetData());
  if (op == Operation::Insert) {
    // 如果是叶节点,插入后 = maxsize就分裂,size <= maxsize - 2
    if (tree_page->IsLeafPage()) {
      return tree_page->GetSize() <= (tree_page->GetMaxSize() - 2);
    }
    // 内部节点, size <= maxsize - 1
    return tree_page->GetSize() <= (tree_page->GetMaxSize() - 1);
  } 
  // 半满即可
  if (op == Operation::Remove) {
    if (tree_page->IsRootPage()) {
      if (tree_page->IsLeafPage()) {
        return tree_page->GetSize() > 0;
      }
      return tree_page->GetSize() > 1; 
    }
    return tree_page->GetSize() > tree_page->GetMinSize();
  }
  return false;
}

INDEX_TEMPLATE_ARGUMENTS
auto BPLUSTREE_TYPE::ReleaseAllParentLatch(Transaction *transaction, bool is_dirty) -> void {
  if (transaction == nullptr) {
    return;
  }
  auto page_set = transaction->GetPageSet();
  while (!page_set->empty()) {
    Page *page = page_set->back();
    page_set->pop_back();
    if (page == nullptr) {
      root_latch_.WUnlock();
    } else {
      page->WUnlatch();
      buffer_pool_manager_->UnpinPage(page->GetPageId(), is_dirty);
    }
  }
}

INDEX_TEMPLATE_ARGUMENTS
auto BPLUSTREE_TYPE::GetLeafPage(Operation op, const KeyType &key, Transaction *transaction) -> LeafPage * {
  if (IsEmpty()) {
    return nullptr;
  }
  page_id_t next_page_id = root_page_id_;
  Page *pre_page = nullptr;
  while (true) {
    Page *fetch_page = buffer_pool_manager_->FetchPage(next_page_id);
    assert(fetch_page != nullptr);
    // crabing ：读，子节点中直接释放父节点的锁
    if (op == Operation::Read) {
      fetch_page->RLatch();
      if (pre_page == nullptr) {
        root_latch_.RUnlock();
      } else {
        pre_page->RUnlatch();
        buffer_pool_manager_->UnpinPage(pre_page->GetPageId(), false);
      }
    }
    // crabing ：写，如果子节点安全，释放祖先节点
    if (op == Operation::Insert || op == Operation::Remove) {
      fetch_page->WLatch();
      if (IsSafePage(fetch_page, op)) {
        ReleaseAllParentLatch(transaction, false);
      }
      transaction->AddIntoPageSet(fetch_page);
    }
    auto tree_page = reinterpret_cast<BPlusTreePage *>(fetch_page->GetData());
    // if is leafPage
    if (tree_page->IsLeafPage()) {
      auto leaf_page = static_cast<LeafPage *>(tree_page);
      transaction->AddIntoPageSet(fetch_page);
      return leaf_page;
    }
    // convert to internale page
    auto internal_page = reinterpret_cast<InternalPage *>(tree_page);
    // 二分查找
    assert(internal_page->GetSize() >= 1);
    int left = 1;
    int right = internal_page->GetSize() - 1;
    int mid;
    while (left <= right) {
      mid = (left + right) / 2;
      if (comparator_(internal_page->KeyAt(mid), key) <= 0) {
        left = mid + 1;
      } else {
        right = mid - 1;
      }
    }
    next_page_id = internal_page->ValueAt(left - 1);
    pre_page = fetch_page;
  }
}

/*****************************************************************************
 * INSERTION
 *****************************************************************************/
/*
 * Insert constant key & value pair into b+ tree
 * if current tree is empty, start new tree, update root page id and insert
 * entry, otherwise insert into leaf page.
 * @return: since we only support unique key, if user try to insert duplicate
 * keys return false, otherwise return true.
 */
INDEX_TEMPLATE_ARGUMENTS
auto BPLUSTREE_TYPE::Insert(const KeyType &key, const ValueType &value, Transaction *transaction) -> bool {
  // std::scoped_lock<std::mutex> lock(latch_);
  // if BPlusTree is empty, init
  root_latch_.WLock();
  if (IsEmpty()) {
    Page *new_page = buffer_pool_manager_->NewPage(&root_page_id_);
    if (new_page == nullptr) {
      root_latch_.WUnlock();
      return false;
    }
    auto leaf_page = reinterpret_cast<LeafPage *>(new_page->GetData());
    leaf_page->Init(root_page_id_, INVALID_PAGE_ID, leaf_max_size_);
    leaf_page->Insert(key, value, comparator_);
    UpdateRootPageId(1);
    root_latch_.WUnlock();
    buffer_pool_manager_->UnpinPage(root_page_id_, true);
    return true;
  }
  transaction->AddIntoPageSet(nullptr);
  // leaf_page会被加写锁，沿途的节点也可能被上写锁
  LeafPage *leaf_page = GetLeafPage(Operation::Insert, key, transaction);
  // if key exists already
  if (!leaf_page->Insert(key, value, comparator_)) {
    // 释放所有资源
    ReleaseAllParentLatch(transaction, false);
    // buffer_pool_manager_->UnpinPage(leaf_page->GetPageId(), false);
    return false;
  }
  // if leaf_page is full, spilt
  if (leaf_page->GetSize() >= leaf_page->GetMaxSize()) {
    Spilt(leaf_page->GetPageId(), transaction);
  }
  ReleaseAllParentLatch(transaction, true);
  return true;
}

INDEX_TEMPLATE_ARGUMENTS
auto BPLUSTREE_TYPE::Spilt(page_id_t page_id, Transaction *transaction) -> void {
  auto leaf_page = reinterpret_cast<LeafPage *>(buffer_pool_manager_->FetchPage(page_id)->GetData());
  page_id_t new_page_id;
  Page *new_page = buffer_pool_manager_->NewPage(&new_page_id);
  assert(new_page != nullptr);
  auto new_leaf_page = reinterpret_cast<LeafPage *>(new_page->GetData());
  new_leaf_page->Init(new_page_id, INVALID_PAGE_ID, leaf_max_size_);
  // 设置next_page_id，便于遍历查找
  page_id_t old_next_page_id = leaf_page->GetNextPageId();
  leaf_page->SetNextPageId(new_page_id);
  new_leaf_page->SetNextPageId(old_next_page_id);
  int size = leaf_page->GetSize();
  int left_num = size / 2;
  // 原leaf_page保留size/2数据，其余插到新leaf_page
  for (int i = left_num; i < size; i++) {
    new_leaf_page->Insert(leaf_page->KeyAt(i), leaf_page->ValueAt(i), comparator_);
  }
  // 修改原leaf_page的size即可，其余数据会被覆盖
  leaf_page->SetSize(left_num);
  // 保存page_id, Spilt_key
  KeyType spilt_key = new_leaf_page->KeyAt(0);
  page_id_t parent_page_id = leaf_page->GetParentPageId();
  BPlusTreePage *left_page = leaf_page;
  BPlusTreePage *right_page = new_leaf_page;
  // 向上递归更新索引
  while (true) {
    // 分裂的为根节点
    if (parent_page_id == INVALID_PAGE_ID) {
      page_id_t root_id;
      Page *new_page = buffer_pool_manager_->NewPage(&root_id);
      assert(new_page != nullptr);
      root_page_id_ = root_id;
      UpdateRootPageId();
      // 设为两个分裂leaf_page的父节点
      left_page->SetParentPageId(root_id);
      right_page->SetParentPageId(root_id);
      // 释放spilt page
      page_id_t left_page_id = left_page->GetPageId();
      page_id_t right_page_id = right_page->GetPageId();
      buffer_pool_manager_->UnpinPage(left_page_id, true);
      buffer_pool_manager_->UnpinPage(right_page_id, true);
      // 初始化内部节点
      auto new_internal_page = reinterpret_cast<InternalPage *>(new_page->GetData());
      new_internal_page->Init(root_id, INVALID_PAGE_ID, internal_max_size_);
      // 更新索引
      new_internal_page->Insert(spilt_key, right_page_id, comparator_);
      new_internal_page->SetValueAt(0, left_page_id);
      // 释放root page
      buffer_pool_manager_->UnpinPage(root_id, true);
      return;
    }
    Page *page = buffer_pool_manager_->FetchPage(parent_page_id);
    std::cout << "parent_page_id: " << parent_page_id << std::endl;
    assert(page != nullptr);
    auto internal_page = reinterpret_cast<InternalPage *>(page->GetData());
    int size = internal_page->GetSize();
    int maxsize = internal_page->GetMaxSize();
    // 内部节点满了, 分裂
    if (size == maxsize) {
      // 先将spilt_key插入已满的内部节点，方便分裂, 更新(maxsize + 1) / 2作为新的spilt_key
      internal_page->Insert(spilt_key, right_page->GetPageId(), comparator_);
      spilt_key = internal_page->KeyAt((maxsize + 1) / 2);
      size += 1;
      // 设置right_page父节点
      right_page->SetParentPageId(parent_page_id);
      // 释放两个page
      buffer_pool_manager_->UnpinPage(left_page->GetPageId(), true);
      buffer_pool_manager_->UnpinPage(right_page->GetPageId(), true);
      // 创建新的internal_page存储KeyAt(maxsize/2+1) - KeyAt(maxsize)
      page_id_t new_page_id;
      Page *new_page = buffer_pool_manager_->NewPage(&new_page_id);
      assert(new_page != nullptr);
      auto new_internal_page = reinterpret_cast<InternalPage *>(new_page->GetData());
      new_internal_page->Init(new_page_id, INVALID_PAGE_ID, internal_max_size_);
      // 将maxsize/2以后的值插入到新page
      new_internal_page->SetValueAt(0, internal_page->ValueAt((maxsize + 1) / 2));
      for (int i = (maxsize + 1) / 2 + 1; i < size; i++) {
        new_internal_page->Insert(internal_page->KeyAt(i), internal_page->ValueAt(i), comparator_);
      }
      internal_page->SetSize((maxsize + 1) / 2);
      // 更新new_internal_page的子指针
      for (int i = 0; i < new_internal_page->GetSize(); i++) {
        Page *page = buffer_pool_manager_->FetchPage(new_internal_page->ValueAt(i));
        assert(page != nullptr);
        auto tree_page = reinterpret_cast<BPlusTreePage *>(page->GetData());
        tree_page->SetParentPageId(new_page_id);
        buffer_pool_manager_->UnpinPage(tree_page->GetPageId(), true);
      }
      // 更新递归参数
      // spilt_key = internal_page->KeyAt(maxsize / 2);
      parent_page_id = internal_page->GetParentPageId();
      left_page = internal_page;
      right_page = new_internal_page;
    } else {
      internal_page->Insert(spilt_key, right_page->GetPageId(), comparator_);
      right_page->SetParentPageId(parent_page_id);
      // 释放两个page
      buffer_pool_manager_->UnpinPage(left_page->GetPageId(), true);
      buffer_pool_manager_->UnpinPage(right_page->GetPageId(), true);
      // 释放parent_page
      buffer_pool_manager_->UnpinPage(internal_page->GetPageId(), true);
      return;
    }
  }
}

INDEX_TEMPLATE_ARGUMENTS
auto BPLUSTREE_TYPE::HandleUnderFlow(BPlusTreePage *tree_page) -> void {
  // 根节点下溢出
  if (tree_page->IsRootPage()) {
    // 根节点不需要考虑半满，若是leaf_page直接返回，internal_page不空即可
    if (tree_page->IsLeafPage() || tree_page->GetSize() > 1) {
      buffer_pool_manager_->UnpinPage(tree_page->GetPageId(), false);
      return;
    }
    // 根节点是内部节点，且无key值，直接释放，树的高度降1
    auto old_internal_page = static_cast<InternalPage *>(tree_page);
    // 将原root_page唯一的一个value作为新的root_page,并且释放原root_page
    root_page_id_ = old_internal_page->ValueAt(0);
    buffer_pool_manager_->UnpinPage(old_internal_page->GetPageId(), true);
    auto internal_page = reinterpret_cast<InternalPage *>(buffer_pool_manager_->FetchPage(root_page_id_)->GetData());
    internal_page->SetParentPageId(INVALID_PAGE_ID);
    UpdateRootPageId();
    buffer_pool_manager_->UnpinPage(root_page_id_, true);
    return;
  }
  // 1. 先向兄弟节点借
  page_id_t sibling_left_id;
  page_id_t sibling_right_id;
  // 获取左右兄弟节点的page_id
  GetSiblings(tree_page, sibling_left_id, sibling_right_id);
  BPlusTreePage *sibling_left_page = nullptr;
  BPlusTreePage *sibling_right_page = nullptr;

  if (sibling_left_id != INVALID_PAGE_ID) {
    sibling_left_page = reinterpret_cast<BPlusTreePage *>(buffer_pool_manager_->FetchPage(sibling_left_id)->GetData());
  }
  if (sibling_right_id != INVALID_PAGE_ID) {
    sibling_right_page =
        reinterpret_cast<BPlusTreePage *>(buffer_pool_manager_->FetchPage(sibling_right_id)->GetData());
  }
  // 获取下溢出节点的父节点
  auto parent_page =
      reinterpret_cast<InternalPage *>(buffer_pool_manager_->FetchPage(tree_page->GetParentPageId())->GetData());

  if (TryBorrow(tree_page, sibling_left_page, parent_page, true)) {
    buffer_pool_manager_->UnpinPage(sibling_left_id, true);
    if (sibling_right_id != INVALID_PAGE_ID) {
      buffer_pool_manager_->UnpinPage(sibling_right_id, false);
    }
    // 释放tree_page,parent_page
    buffer_pool_manager_->UnpinPage(parent_page->GetPageId(), true);
    buffer_pool_manager_->UnpinPage(tree_page->GetPageId(), true);
    return;
  }
  if (TryBorrow(tree_page, sibling_right_page, parent_page, false)) {
    buffer_pool_manager_->UnpinPage(sibling_right_id, true);
    if (sibling_left_id != INVALID_PAGE_ID) {
      buffer_pool_manager_->UnpinPage(sibling_left_id, false);
    }
    // 释放tree_page,parent_page
    buffer_pool_manager_->UnpinPage(parent_page->GetPageId(), true);
    buffer_pool_manager_->UnpinPage(tree_page->GetPageId(), true);
    return;
  }
  // fail to borrow from left and right page, try to merge page
  if (sibling_left_page != nullptr) {
    Merge(sibling_left_page, tree_page, parent_page);
    buffer_pool_manager_->UnpinPage(sibling_left_id, true);
    if (sibling_right_page != nullptr) {
      buffer_pool_manager_->UnpinPage(sibling_right_id, false);
    }
  } else if (sibling_right_page != nullptr) {
    Merge(tree_page, sibling_right_page, parent_page);
    buffer_pool_manager_->UnpinPage(sibling_right_id, true);
  } else {
    throw std::invalid_argument("unroot page did not have left or right brother page!");
  }
  buffer_pool_manager_->UnpinPage(tree_page->GetPageId(), true);
  // 父节点下溢出，递归处理
  if (parent_page->GetSize() < parent_page->GetMinSize()) {
    HandleUnderFlow(parent_page);
  } else {
    buffer_pool_manager_->UnpinPage(parent_page->GetPageId(), true);
  }
}
INDEX_TEMPLATE_ARGUMENTS
auto BPLUSTREE_TYPE::GetSiblings(BPlusTreePage *tree_page, page_id_t &sibling_left_id, page_id_t &sibling_right_id)
    -> void {
  if (tree_page->IsRootPage()) {
    throw std::invalid_argument("can not sibling from root page\n");
  }
  auto parent_page =
      reinterpret_cast<InternalPage *>(buffer_pool_manager_->FetchPage(tree_page->GetParentPageId())->GetData());
  sibling_left_id = sibling_right_id = INVALID_PAGE_ID;
  int index = parent_page->GetdValueIndex(tree_page->GetPageId());
  if (index == -1) {
    throw std::logic_error("child page did not exists in parent page");
  }
  if (index != 0) {
    sibling_left_id = parent_page->ValueAt(index - 1);
  }
  if (index != parent_page->GetSize() - 1) {
    sibling_right_id = parent_page->ValueAt(index + 1);
  }
  buffer_pool_manager_->UnpinPage(tree_page->GetParentPageId(), false);
}

INDEX_TEMPLATE_ARGUMENTS
auto BPLUSTREE_TYPE::TryBorrow(BPlusTreePage *tree_page, BPlusTreePage *sibling_page, InternalPage *parent_page,
                               bool is_left) -> bool {
  if (sibling_page == nullptr || sibling_page->GetSize() <= sibling_page->GetMinSize()) {
    return false;
  }
  // 左边兄弟借最右边，右边兄弟leaf_page借第一个/internal_page借第二个
  int borrow_index = is_left ? (sibling_page->GetSize() - 1) : ((sibling_page->IsLeafPage()) ? 0 : 1);
  // 借右边，更新父节点指向sibling_page的指针，借左边，更新父节点指向tree_page的指针
  int page_index = parent_page->GetdValueIndex(tree_page->GetPageId());
  int update_parent_index = is_left ? page_index : page_index + 1;
  // 如果是leaf_page, 从左边借，更新的值就是借来的值，从右边借，更新的值是借来的下一个值
  KeyType update_key;
  if (tree_page->IsLeafPage()) {
    auto leaf_page = static_cast<LeafPage *>(tree_page);
    auto leaf_sibling_page = static_cast<LeafPage *>(sibling_page);
    leaf_page->Insert(leaf_sibling_page->KeyAt(borrow_index), leaf_sibling_page->ValueAt(borrow_index), comparator_);
    leaf_sibling_page->Remove(leaf_sibling_page->KeyAt(borrow_index), comparator_);
    update_key = is_left ? leaf_page->KeyAt(0) : leaf_sibling_page->KeyAt(0);
  } else {
    auto internal_page = static_cast<InternalPage *>(tree_page);
    auto internal_sibling_page = static_cast<InternalPage *>(sibling_page);
    update_key = internal_sibling_page->KeyAt(borrow_index);
    page_id_t child_id = INVALID_PAGE_ID;
    // 如果是向左借
    if (is_left) {
      internal_page->Insert(parent_page->KeyAt(update_parent_index), internal_page->ValueAt(0), comparator_);
      internal_page->SetValueAt(0, internal_sibling_page->ValueAt(borrow_index));
      // 删除左边左后一个节点
      internal_sibling_page->DecreaseSize(1);
      child_id = internal_page->ValueAt(0);
    } else {
      internal_page->Insert(parent_page->KeyAt(update_parent_index), internal_sibling_page->ValueAt(0), comparator_);
      // 删除右边第一个节点
      internal_sibling_page->RemoveAt(0);
      child_id = internal_page->ValueAt(internal_page->GetSize() - 1);
    }
    // 更新移动的child_page的parent_page_id
    auto tree_page = reinterpret_cast<BPlusTreePage *>(buffer_pool_manager_->FetchPage(child_id)->GetData());
    tree_page->SetParentPageId(internal_page->GetPageId());
    buffer_pool_manager_->UnpinPage(child_id, true);
  }
  // 更新父节点的值
  parent_page->SetKeyAt(update_parent_index, update_key);
  return true;
}

INDEX_TEMPLATE_ARGUMENTS
auto BPLUSTREE_TYPE::Merge(BPlusTreePage *left_page, BPlusTreePage *right_page, InternalPage *parent_page) -> void {
  if (left_page->IsLeafPage()) {
    auto leaf_left_page = static_cast<LeafPage *>(left_page);
    auto leaf_right_page = static_cast<LeafPage *>(right_page);
    // 将右边合并到左边
    for (int i = 0; i < leaf_right_page->GetSize(); i++) {
      leaf_left_page->Insert(leaf_right_page->KeyAt(i), leaf_right_page->ValueAt(i), comparator_);
    }
    // 更新next_page_id
    page_id_t next_page_id = leaf_right_page->GetNextPageId();
    leaf_left_page->SetNextPageId(next_page_id);
    // 删除父节点索引
    int index = parent_page->GetdValueIndex(leaf_right_page->GetPageId());
    parent_page->RemoveAt(index);
  } else {
    auto internal_left_page = static_cast<InternalPage *>(left_page);
    auto internal_right_page = static_cast<InternalPage *>(right_page);
    // 将父节点索引移动到internal_left_page中
    int index = parent_page->GetdValueIndex(internal_right_page->GetPageId());
    internal_right_page->SetKeyAt(0, parent_page->KeyAt(index));
    // internal_left_page->Insert(parent_page->KeyAt(index), internal_right_page->ValueAt(0), comparator_);
    // SetPageParent(internal_right_page->ValueAt(0), internal_left_page->GetPageId());
    parent_page->RemoveAt(index);
    for (int i = 0; i < internal_right_page->GetSize(); i++) {
      internal_left_page->Insert(internal_right_page->KeyAt(i), internal_right_page->ValueAt(i), comparator_);
      SetPageParent(internal_right_page->ValueAt(i), internal_left_page->GetPageId());
    }
  }
}

INDEX_TEMPLATE_ARGUMENTS
auto BPLUSTREE_TYPE::SetPageParent(page_id_t page_id, page_id_t parent_id) -> void {
  auto tree_page = reinterpret_cast<BPlusTreePage *>(buffer_pool_manager_->FetchPage(page_id)->GetData());
  tree_page->SetParentPageId(parent_id);
  buffer_pool_manager_->UnpinPage(page_id, true);
}

/*****************************************************************************
 * REMOVE
 *****************************************************************************/
/*
 * Delete key & value pair associated with input key
 * If current tree is empty, return immdiately.
 * If not, User needs to first find the right leaf page as deletion target, then
 * delete entry from leaf page. Remember to deal with redistribute or merge if
 * necessary.
 */
INDEX_TEMPLATE_ARGUMENTS
void BPLUSTREE_TYPE::Remove(const KeyType &key, Transaction *transaction) {
  std::scoped_lock<std::mutex> lock(latch_);
  if (IsEmpty()) {
    return;
  }
  // 找到key所在leaf_page
  LeafPage *leaf_page = GetLeafPage(Operation::Remove, key, transaction);
  // 不存在该key
  if (!leaf_page->Remove(key, comparator_)) {
    buffer_pool_manager_->UnpinPage(leaf_page->GetPageId(), false);
    return;
  }
  // 删除成功，叶子节点是根节点，不需要判断下溢出，直接返回
  if (leaf_page->IsRootPage()) {
    buffer_pool_manager_->UnpinPage(leaf_page->GetPageId(), true);
    return;
  }
  // leaf_paeg下溢出，向上递归处理
  if (leaf_page->GetSize() < leaf_page->GetMinSize()) {
    HandleUnderFlow(leaf_page);
  } else {
    buffer_pool_manager_->UnpinPage(leaf_page->GetPageId(), true);
  }
}

/*****************************************************************************
 * INDEX ITERATOR
 *****************************************************************************/
/*
 * Input parameter is void, find the leaftmost leaf page first, then construct
 * index iterator
 * @return : index iterator
 */
INDEX_TEMPLATE_ARGUMENTS
auto BPLUSTREE_TYPE::Begin() -> INDEXITERATOR_TYPE {
  // BPlusTree is empty
  if (IsEmpty()) {
    return INDEXITERATOR_TYPE();
  }
  auto next_page_id = root_page_id_;
  while (true) {
    auto page = buffer_pool_manager_->FetchPage(next_page_id);
    assert(page != nullptr);
    auto tree_page = reinterpret_cast<BPlusTreePage *>(page->GetData());
    // construct iterator
    if (tree_page->IsLeafPage()) {
      return INDEXITERATOR_TYPE(tree_page, buffer_pool_manager_);
    }
    auto internal_page = static_cast<InternalPage *>(tree_page);
    next_page_id = internal_page->ValueAt(0);
  }
}

/*
 * Input parameter is low key, find the leaf page that contains the input key
 * first, then construct index iterator
 * @return : index iterator
 */
INDEX_TEMPLATE_ARGUMENTS
auto BPLUSTREE_TYPE::Begin(const KeyType &key) -> INDEXITERATOR_TYPE {
  LeafPage *leaf_page = GetLeafPage(key);
  if (leaf_page == nullptr) {
    return INDEXITERATOR_TYPE();
  }
  int size = leaf_page->GetSize();
  for (int i = 0; i < size; i++) {
    if (comparator_(leaf_page->KeyAt(i), key) == 0) {
      return INDEXITERATOR_TYPE(leaf_page, buffer_pool_manager_, i);
    }
  }
  // did not find key
  return INDEXITERATOR_TYPE();
}

/*
 * Input parameter is void, construct an index iterator representing the end
 * of the key/value pair in the leaf node
 * @return : index iterator
 */
INDEX_TEMPLATE_ARGUMENTS
auto BPLUSTREE_TYPE::End() -> INDEXITERATOR_TYPE { return INDEXITERATOR_TYPE(); }

/**
 * @return Page id of the root of this tree
 */
INDEX_TEMPLATE_ARGUMENTS
auto BPLUSTREE_TYPE::GetRootPageId() -> page_id_t { return root_page_id_; }

/*****************************************************************************
 * UTILITIES AND DEBUG
 *****************************************************************************/
/*
 * Update/Insert root page id in header page(where page_id = 0, header_page is
 * defined under include/page/header_page.h)
 * Call this method everytime root page id is changed.
 * @parameter: insert_record      defualt value is false. When set to true,
 * insert a record <index_name, root_page_id> into header page instead of
 * updating it.
 */
INDEX_TEMPLATE_ARGUMENTS
void BPLUSTREE_TYPE::UpdateRootPageId(int insert_record) {
  auto *header_page = static_cast<HeaderPage *>(buffer_pool_manager_->FetchPage(HEADER_PAGE_ID));
  if (insert_record != 0) {
    // create a new record<index_name + root_page_id> in header_page
    header_page->InsertRecord(index_name_, root_page_id_);
  } else {
    // update root_page_id in header_page
    header_page->UpdateRecord(index_name_, root_page_id_);
  }
  buffer_pool_manager_->UnpinPage(HEADER_PAGE_ID, true);
}

/*
 * This method is used for test only
 * Read data from file and insert one by one
 */
INDEX_TEMPLATE_ARGUMENTS
void BPLUSTREE_TYPE::InsertFromFile(const std::string &file_name, Transaction *transaction) {
  int64_t key;
  std::ifstream input(file_name);
  while (input) {
    input >> key;

    KeyType index_key;
    index_key.SetFromInteger(key);
    RID rid(key);
    Insert(index_key, rid, transaction);
  }
}
/*
 * This method is used for test only
 * Read data from file and remove one by one
 */
INDEX_TEMPLATE_ARGUMENTS
void BPLUSTREE_TYPE::RemoveFromFile(const std::string &file_name, Transaction *transaction) {
  int64_t key;
  std::ifstream input(file_name);
  while (input) {
    input >> key;
    KeyType index_key;
    index_key.SetFromInteger(key);
    Remove(index_key, transaction);
  }
}

/**
 * This method is used for debug only, You don't need to modify
 */
INDEX_TEMPLATE_ARGUMENTS
void BPLUSTREE_TYPE::Draw(BufferPoolManager *bpm, const std::string &outf) {
  if (IsEmpty()) {
    LOG_WARN("Draw an empty tree");
    return;
  }
  std::ofstream out(outf);
  out << "digraph G {" << std::endl;
  ToGraph(reinterpret_cast<BPlusTreePage *>(bpm->FetchPage(root_page_id_)->GetData()), bpm, out);
  out << "}" << std::endl;
  out.flush();
  out.close();
}

/**
 * This method is used for debug only, You don't need to modify
 */
INDEX_TEMPLATE_ARGUMENTS
void BPLUSTREE_TYPE::Print(BufferPoolManager *bpm) {
  if (IsEmpty()) {
    LOG_WARN("Print an empty tree");
    return;
  }
  ToString(reinterpret_cast<BPlusTreePage *>(bpm->FetchPage(root_page_id_)->GetData()), bpm);
}

/**
 * This method is used for debug only, You don't need to modify
 * @tparam KeyType
 * @tparam ValueType
 * @tparam KeyComparator
 * @param page
 * @param bpm
 * @param out
 */
INDEX_TEMPLATE_ARGUMENTS
void BPLUSTREE_TYPE::ToGraph(BPlusTreePage *page, BufferPoolManager *bpm, std::ofstream &out) const {
  std::string leaf_prefix("LEAF_");
  std::string internal_prefix("INT_");
  if (page->IsLeafPage()) {
    auto *leaf = reinterpret_cast<LeafPage *>(page);
    // Print node name
    out << leaf_prefix << leaf->GetPageId();
    // Print node properties
    out << "[shape=plain color=green ";
    // Print data of the node
    out << "label=<<TABLE BORDER=\"0\" CELLBORDER=\"1\" CELLSPACING=\"0\" CELLPADDING=\"4\">\n";
    // Print data
    out << "<TR><TD COLSPAN=\"" << leaf->GetSize() << "\">P=" << leaf->GetPageId() << "</TD></TR>\n";
    out << "<TR><TD COLSPAN=\"" << leaf->GetSize() << "\">"
        << "max_size=" << leaf->GetMaxSize() << ",min_size=" << leaf->GetMinSize() << ",size=" << leaf->GetSize()
        << "</TD></TR>\n";
    out << "<TR>";
    for (int i = 0; i < leaf->GetSize(); i++) {
      out << "<TD>" << leaf->KeyAt(i) << "</TD>\n";
    }
    out << "</TR>";
    // Print table end
    out << "</TABLE>>];\n";
    // Print Leaf node link if there is a next page
    if (leaf->GetNextPageId() != INVALID_PAGE_ID) {
      out << leaf_prefix << leaf->GetPageId() << " -> " << leaf_prefix << leaf->GetNextPageId() << ";\n";
      out << "{rank=same " << leaf_prefix << leaf->GetPageId() << " " << leaf_prefix << leaf->GetNextPageId() << "};\n";
    }

    // Print parent links if there is a parent
    if (leaf->GetParentPageId() != INVALID_PAGE_ID) {
      out << internal_prefix << leaf->GetParentPageId() << ":p" << leaf->GetPageId() << " -> " << leaf_prefix
          << leaf->GetPageId() << ";\n";
    }
  } else {
    auto *inner = reinterpret_cast<InternalPage *>(page);
    // Print node name
    out << internal_prefix << inner->GetPageId();
    // Print node properties
    out << "[shape=plain color=pink ";  // why not?
    // Print data of the node
    out << "label=<<TABLE BORDER=\"0\" CELLBORDER=\"1\" CELLSPACING=\"0\" CELLPADDING=\"4\">\n";
    // Print data
    out << "<TR><TD COLSPAN=\"" << inner->GetSize() << "\">P=" << inner->GetPageId() << "</TD></TR>\n";
    out << "<TR><TD COLSPAN=\"" << inner->GetSize() << "\">"
        << "max_size=" << inner->GetMaxSize() << ",min_size=" << inner->GetMinSize() << ",size=" << inner->GetSize()
        << "</TD></TR>\n";
    out << "<TR>";
    for (int i = 0; i < inner->GetSize(); i++) {
      out << "<TD PORT=\"p" << inner->ValueAt(i) << "\">";
      if (i > 0) {
        out << inner->KeyAt(i);
      } else {
        out << " ";
      }
      out << "</TD>\n";
    }
    out << "</TR>";
    // Print table end
    out << "</TABLE>>];\n";
    // Print Parent link
    if (inner->GetParentPageId() != INVALID_PAGE_ID) {
      out << internal_prefix << inner->GetParentPageId() << ":p" << inner->GetPageId() << " -> " << internal_prefix
          << inner->GetPageId() << ";\n";
    }
    // Print leaves
    for (int i = 0; i < inner->GetSize(); i++) {
      auto child_page = reinterpret_cast<BPlusTreePage *>(bpm->FetchPage(inner->ValueAt(i))->GetData());
      ToGraph(child_page, bpm, out);
      if (i > 0) {
        auto sibling_page = reinterpret_cast<BPlusTreePage *>(bpm->FetchPage(inner->ValueAt(i - 1))->GetData());
        if (!sibling_page->IsLeafPage() && !child_page->IsLeafPage()) {
          out << "{rank=same " << internal_prefix << sibling_page->GetPageId() << " " << internal_prefix
              << child_page->GetPageId() << "};\n";
        }
        bpm->UnpinPage(sibling_page->GetPageId(), false);
      }
    }
  }
  bpm->UnpinPage(page->GetPageId(), false);
}

/**
 * This function is for debug only, you don't need to modify
 * @tparam KeyType
 * @tparam ValueType
 * @tparam KeyComparator
 * @param page
 * @param bpm
 */
INDEX_TEMPLATE_ARGUMENTS
void BPLUSTREE_TYPE::ToString(BPlusTreePage *page, BufferPoolManager *bpm) const {
  if (page->IsLeafPage()) {
    auto *leaf = reinterpret_cast<LeafPage *>(page);
    std::cout << "Leaf Page: " << leaf->GetPageId() << " parent: " << leaf->GetParentPageId()
              << " next: " << leaf->GetNextPageId() << std::endl;
    for (int i = 0; i < leaf->GetSize(); i++) {
      std::cout << leaf->KeyAt(i) << ",";
    }
    std::cout << std::endl;
    std::cout << std::endl;
  } else {
    auto *internal = reinterpret_cast<InternalPage *>(page);
    std::cout << "Internal Page: " << internal->GetPageId() << " parent: " << internal->GetParentPageId() << std::endl;
    for (int i = 0; i < internal->GetSize(); i++) {
      std::cout << internal->KeyAt(i) << ": " << internal->ValueAt(i) << ",";
    }
    std::cout << std::endl;
    std::cout << std::endl;
    for (int i = 0; i < internal->GetSize(); i++) {
      ToString(reinterpret_cast<BPlusTreePage *>(bpm->FetchPage(internal->ValueAt(i))->GetData()), bpm);
    }
  }
  bpm->UnpinPage(page->GetPageId(), false);
}

template class BPlusTree<GenericKey<4>, RID, GenericComparator<4>>;
template class BPlusTree<GenericKey<8>, RID, GenericComparator<8>>;
template class BPlusTree<GenericKey<16>, RID, GenericComparator<16>>;
template class BPlusTree<GenericKey<32>, RID, GenericComparator<32>>;
template class BPlusTree<GenericKey<64>, RID, GenericComparator<64>>;

}  // namespace bustub
